package com.qf.flinketl.mapper;

import com.qf.flinketl.flinkconfig.MapperConfig;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serial;
import java.io.Serializable;

/**
 * @Description: @Author: KyrieFc @Date: 2023/10/22
 */
public class UserMapper extends RichFlatMapFunction<SourceMessage, MessageRow>
    implements Serializable {
  @Serial private static final long serialVersionUID = 1L;
  private static final Logger logger = LoggerFactory.getLogger(UserMapper.class);
  private MapperConfig mapperConfig = null;

  @Override
  public void open(Configuration parameters) throws Exception {
    super.open(parameters);
    logger.info("UserMapper open");
  }

  @Override
  public void flatMap(SourceMessage sourceMessage, Collector<MessageRow> collector)
      throws Exception {
    try {
      long currentTimeMillis = System.currentTimeMillis();
      String message = sourceMessage.getMessage().toString();
      MessageRow messageRow = new MessageRow();
      messageRow.setMessage(message);
      collector.collect(messageRow);
    } catch (Exception e) {
      logger.error("UserMapper error: {}", e.getMessage());
    }
  }

  @Override
  public void close() throws Exception {
    super.close();
  }

  public MapperConfig getMapperConfig() {
    return mapperConfig;
  }

  public void setMapperConfig(MapperConfig mapperConfig) {
    this.mapperConfig = mapperConfig;
  }
}
